package com.hadoop.mr.inputformat;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapred.FileSplit;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import java.io.IOException;

/**
 * @program: hadoop
 * @description: 自定义ReacordReader
 * @author: Qiang.Ye
 * @date: 2019-10-24 06:45
 **/
public class PlusRecordReader extends RecordReader<NullWritable, BytesWritable> {

    private FileSplit inputSplit;
    private Configuration configuration;

    //是否重复读
    private  boolean progressd = false;
    private  BytesWritable value = new BytesWritable();

    @Override
    public void initialize(InputSplit inputSplit, TaskAttemptContext context) throws IOException, InterruptedException {
        this.inputSplit = (FileSplit) inputSplit;
        this.configuration = context.getConfiguration();
    }

    @Override
    public boolean nextKeyValue() throws IOException, InterruptedException {
        if(!progressd) {
            FSDataInputStream fis = null;
            try {
              //定义一个存放数据的缓冲区
              byte[] contents = new byte[(int) inputSplit.getLength()];
              //获取文件系统
              Path path = inputSplit.getPath();
              FileSystem fs = path.getFileSystem(configuration);
              //读取数据
              fis = fs.open(path);
              //将文件内容写入到缓冲区
              IOUtils.readFully(fis, contents, 0, contents.length);
              //文件内容输出
              value.set(contents, 0, contents.length);
              //判断是否读取结束

          }catch (Exception e){
            e.printStackTrace();
          }finally {
                IOUtils.closeStream(fis);
          }
          progressd = true;
          return true;
        }
        return false;
    }

    @Override
    public NullWritable getCurrentKey() throws IOException, InterruptedException {
        return null;
    }

    @Override
    public BytesWritable getCurrentValue() throws IOException, InterruptedException {
        return null;
    }

    @Override
    public float getProgress() throws IOException, InterruptedException {
        return 0;
    }

    @Override
    public void close() throws IOException {

    }
}
